package com.lzy.gmall.realtime.app.dwd;

import com.lzy.gmall.realtime.utils.MyKafkaUtil;
import com.lzy.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/*
*
*                                      交易域订单明细事实表
*
* */
public class Dwd_Transaction_field_order_details_Table {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        1. 创建执行环境并设置并行度
        env.setParallelism(1);

        // 创建表环境（用于Flink SQL操作）
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // 设置状态过期时间（5秒未更新则清理，避免内存溢出）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
//        4. 从Kafka读取业务数据库的binlog日志
        tableEnv.executeSql(MyKafkaUtil.getTopicDb("order_pre_process_211126"));

//      订单表
        Table order_info = tableEnv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount ,\n" +
                "`data`['order_status'] order_status ,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['delivery_address'] delivery_address,\n" +
                "`data`['order_comment'] order_comment,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['process_status'] process_status,\n" +
                "`data`['tracking_no'] tracking_no,\n" +
                "`data`['parent_order_id'] parent_order_id,\n" +
                "`data`['img_url'] img_url,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['feight_fee'] feight_fee,\n" +
                "`data`['feight_fee_reduce'] feight_fee_reduce,\n" +
                "`data`['refundable_time'] refundable_time,\n" +
                " `type`, \n" +
                " `old` \n" +
                " from topic_db " +
                "where `database`='gmall' " +
                "and `table`='order_info' " +
                "and (`type`='insert' or `type`='update') ");
        tableEnv.createTemporaryView("order_info",order_info);


//      订单明细表
        Table order_detail = tableEnv.sqlQuery(" select\n" +
                "`data`['id'] id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['sku_name'] sku_name ,\n" +
                "`data`['img_url'] img_url ,\n" +
                "`data`['order_price'] order_price,\n" +
                "`data`['sku_num'] sku_num,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['source_type'] source_type,\n" +
                "`data`['source_id'] source_id,\n" +
                "`data`['split_total_amount'] split_total_amount,\n" +
                "`data`['split_activity_amount'] split_activity_amount,\n" +
                "`data`['split_coupon_amount'] split_coupon_amount,\n" +
                "pt \n" +
                " from topic_db " +
                "where `database`='gmall' " +
                "and `table`='order_detail' " +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_detail",order_detail);



        //      用户信息表
        Table user_info = tableEnv.sqlQuery("SELECT\n" +
                "    `data`['id'] id,\n" +
                "    `data`['login_name'] login_name,\n" +
                "    `data`['nick_name'] nick_name,\n" +
                "    `data`['name'] name,\n" +
                "    `data`['phone_num'] phone_num,\n" +
                "    `data`['email'] email,\n" +
                "    `data`['user_level'] user_level,\n" +
                "    `data`['birthday'] birthday,\n" +
                "    `data`['gender'] gender,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['operate_time'] operate_time,\n" +
                "    `data`['status'] status\n" +
                "FROM topic_db\n" +
                "WHERE `database` = 'gmall'\n" +
                "  AND `table` = 'user_info'\n" +
                "  AND (`type` = 'insert' OR `type` = 'update')");
        tableEnv.createTemporaryView("user_info", user_info);


//      支付信息表
        Table payment_info = tableEnv.sqlQuery("SELECT\n" +
                "    `data`['id'] id,\n" +
                "    `data`['out_trade_no'] out_trade_no,\n" +
                "    `data`['order_id'] order_id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['payment_type'] payment_type,\n" +
                "    `data`['trade_no'] trade_no,\n" +
                "    `data`['total_amount'] total_amount,\n" +
                "    `data`['subject'] subject,\n" +
                "    `data`['payment_status'] payment_status,\n" +
                "    `data`['create_time'] create_time,\n" +
                "    `data`['callback_time'] callback_time,\n" +
                "    `data`['callback_content'] callback_content\n" +
                "FROM topic_db\n" +
                "WHERE `database` = 'gmall'\n" +
                "  AND `table` = 'payment_info'\n" +
                "  AND (`type` = 'insert' OR `type` = 'update')");
        tableEnv.createTemporaryView("payment_info", payment_info);



//      订单详情相关的活动记录表
        Table order_detail_activity = tableEnv.sqlQuery("SELECT\n" +
                "    `data`['id'] id,\n" +
                "    `data`['order_detail_id'] order_detail_id,\n" +
                "    `data`['activity_id'] activity_id,\n" +
                "    `data`['activity_rule_id'] activity_rule_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['create_time'] create_time\n" +
                "FROM topic_db\n" +
                "WHERE `database` = 'gmall'\n" +
                "  AND `table` = 'order_detail_activity'\n" +
                "  AND (`type` = 'insert' OR `type` = 'update')" );
        tableEnv.createTemporaryView("order_detail_activity", order_detail_activity);
//        tableEnv.toRetractStream(order_detail_activity, Row.class).print();
        tableEnv.executeSql(MysqlUtil.getBaseDicLookUpDDL());


        //order_detail+order_info+order_detail_activity+user_info+payment_info+base_dic

        final Table table5 = tableEnv.sqlQuery("select " +
                        "   od.id, " +
                        "   od.order_id, " +
                        "   od.sku_id, " +
                        "   od.sku_name, " +
                        "   od.order_price, " +
                        "   od.sku_num, " +
                        "   od.create_time, " +
                        "   od.source_type source_type_id, " +
                        "   dic.dic_name source_type_name, " +
                        "   od.source_id, " +
                        "   od.split_total_amount, " +
                        "   od.split_activity_amount, " +
                        "   od.split_coupon_amount, " +
                        "   oi.consignee, " +
                        "   oi.consignee_tel, " +
                        "   oi.total_amount, " +
                        "   oi.order_status, " +
                        "   oi.user_id, " +
                        "   oi.payment_way, " +
                        "   oi.delivery_address, " +
                        "   oi.order_comment, " +
                        "   oi.out_trade_no, " +
                        "   oi.trade_body, " +
                        "   oi.operate_time, " +
                        "   oi.expire_time, " +
                        "   oi.process_status, " +
                        "   oi.tracking_no, " +
                        "   oi.parent_order_id, " +
                        "   oi.province_id, " +
                        "   oi.activity_reduce_amount, " +
                        "   oi.coupon_reduce_amount, " +
                        "   oi.original_total_amount, " +
                        "   oi.feight_fee, " +
                        "   oi.feight_fee_reduce, " +
                        "   oi.refundable_time, " +
                        "   oa.id order_detail_activity_id, " +
                        "   oa.activity_id, " +
                        "   oa.activity_rule_id, " +
                        "   oi.`type`, " +
                        "   oi.`old`, " +
                        "current_row_timestamp() row_op_ts " +
                        "from order_detail od \n" +
                        "join order_info oi on od.order_id=oi.id\n" +
                        "left join order_detail_activity oa on oa.order_detail_id=od.id\n" +
                "LEFT JOIN user_info ui on ui.id=oi.user_id\n" +
                        "LEFT JOIN payment_info pi on pi.order_id=oi.id\n" +
                "join base_dic FOR SYSTEM_TIME AS OF od.pt as dic\n" +
                        "on od.source_type=dic.dic_code");

        tableEnv.createTemporaryView("result_table",table5);
//        tableEnv.toRetractStream(table5, Row.class).print();



//              交易域订单明细事实表
//      修改目标表DDL，移除order_detail_coupon_id列
        tableEnv.executeSql("CREATE TABLE dwd_order_pre (\n" +
                "    `id` STRING COMMENT '订单明细ID',\n" +
                "    `order_id` STRING COMMENT '订单ID',\n" +
                "    `sku_id` STRING COMMENT '商品SKU ID',\n" +
                "    `sku_name` STRING COMMENT '商品名称',\n" +
                "    `order_price` STRING COMMENT '订单价格',\n" +
                "    `sku_num` STRING COMMENT '商品数量',\n" +
                "    `create_time` STRING COMMENT '创建时间',\n" +
                "    `source_type_id` STRING COMMENT '来源类型ID',\n" +
                "    `source_type_name` STRING COMMENT '来源类型名称',\n" +
                "    `source_id` STRING COMMENT '来源ID',\n" +
                "    `split_total_amount` STRING COMMENT '分摊总金额',\n" +
                "    `split_activity_amount` STRING COMMENT '分摊活动金额',\n" +
                "    `split_coupon_amount` STRING COMMENT '分摊优惠券金额',\n" +
                "    `consignee` STRING COMMENT '收货人',\n" +
                "    `consignee_tel` STRING COMMENT '收货人电话',\n" +
                "    `total_amount` STRING COMMENT '订单总金额',\n" +
                "    `order_status` STRING COMMENT '订单状态',\n" +
                "    `user_id` STRING COMMENT '用户ID',\n" +
                "    `payment_way` STRING COMMENT '支付方式',\n" +
                "    `delivery_address` STRING COMMENT '配送地址',\n" +
                "    `order_comment` STRING COMMENT '订单评论',\n" +
                "    `out_trade_no` STRING COMMENT '外部交易号',\n" +
                "    `trade_body` STRING COMMENT '交易内容',\n" +
                "    `operate_time` STRING COMMENT '操作时间',\n" +
                "    `expire_time` STRING COMMENT '过期时间',\n" +
                "    `process_status` STRING COMMENT '处理状态',\n" +
                "    `tracking_no` STRING COMMENT '物流单号',\n" +
                "    `parent_order_id` STRING COMMENT '父订单ID',\n" +
                "    `province_id` STRING COMMENT '省份ID',\n" +
                "    `activity_reduce_amount` STRING COMMENT '活动减免金额',\n" +
                "    `coupon_reduce_amount` STRING COMMENT '优惠券减免金额',\n" +
                "    `original_total_amount` STRING COMMENT '原始总金额',\n" +
                "    `feight_fee` STRING COMMENT '运费',\n" +
                "    `feight_fee_reduce` STRING COMMENT '运费减免',\n" +
                "    `refundable_time` STRING COMMENT '可退款时间',\n" +
                "    `order_detail_activity_id` STRING COMMENT '订单明细活动ID',\n" +
                "    `activity_id` STRING COMMENT '活动ID',\n" +
                "    `activity_rule_id` STRING COMMENT '活动规则ID',\n" +
                "    `type` STRING COMMENT '操作类型',\n" +
                "    `old` MAP<STRING, STRING> COMMENT '旧值映射',\n" +
                "    `row_op_ts` TIMESTAMP_LTZ(3) COMMENT '行操作时间戳',\n" +
                "    PRIMARY KEY (`id`) NOT ENFORCED" +
                ")"+MyKafkaUtil.getUpsertKafkaDDL("dwd_order_pre"));

                tableEnv.executeSql("insert into dwd_order_pre select * from result_table");
                tableEnv.executeSql("select * from dwd_order_pre").print();






        env.execute();
    }
}